package org.niit.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Encoder, Encoders, Row, SparkSession, functions}

/*
 UDF              UDAF

 用户自定义函数

 */
object SparkSQL_3 {
  def main(args: Array[String]): Unit = {

/*
    +--------+
    |username|
    +--------+
    |zhangsan|
    |    lisi|
    |  wangwu|
    +--------+

    +--------+
    |username|
    +--------+
    |name:zhangsan|
    |name:lisi|
    |name:wangwu|
    +--------+

    */

    //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")
    //1.DataFrame
    //1.1使用spark SQL 读取 json文件
    val dataFrame: DataFrame = spark.read.json("input/user.json")
    dataFrame.createOrReplaceTempView("user")
    //定义一个 UDF函数  第一个参数 函数名称  第二个参数 参数列表。 传值进行的形参
    spark.udf.register("preName",(name:String)=>{
       "name:" + name//返回
      //name.substring(0,2)
    })

    //调用自定的udf函数
    spark.sql("select preName(username) from user").show()


    //2.UDAF 弱类型
    //spark.udf.register("ageAvg",new MyAvgUDAF)
    //spark.sql("select ageAvg(age) from user ").show


    //2.UDAF 强类型
    spark.udf.register("ageAvg",functions.udaf(new MyAvgUDAF))
    spark.sql("select ageAvg(age) from user ").show

    dataFrame.selectExpr("AVG(age)").show()
    import spark.implicits._
    dataFrame.agg("avg"->"age").show()
    spark.close()

  }

  
    //UDAF 强类型
  //定义缓冲区
  case class Buff(var total:Long,var count:Long)
                                // 输入数据的类型   缓冲区类型        输出数据的类型
  class MyAvgUDAF extends Aggregator[Long,         Buff,            Long]{
      //缓冲区进行初始化
    override def zero: Buff = {
      Buff(0L,0L)
    }
    //根据输入的数据更新缓冲区的数据    age
    override def reduce(buff: Buff, in: Long): Buff = {
      buff.total = buff.total + in
      buff.count = buff.count + 1

      buff
    }
     //合并缓冲区
    override def merge(b1: Buff, b2: Buff): Buff = {
      b1.total = b1.total + b2.total
      b1.count = b1.count + b2.count
      b1
    }
    //计算结果
    override def finish(buff: Buff): Long = {
      buff.total / buff.count
    }
    //缓冲区的编码操作
    override def bufferEncoder: Encoder[Buff] = Encoders.product
    //输出的编码操作
    override def outputEncoder: Encoder[Long] = Encoders.scalaLong
    }




  // UDAF 弱类型  理解  计算年龄的平均值
  /*class MyAvgUDAF extends UserDefinedAggregateFunction{
    // 输入数据的结构 Long
    override def inputSchema: StructType = {
      StructType(
        Array(
          StructField("age",LongType)

        )
      )

    }

    //缓冲区的数据结构 ：Buffer
    override def bufferSchema: StructType = {
      StructType(
        Array(
          StructField("total",LongType),//年龄的总和
          StructField("count",LongType)//年龄出现的次数
        )
      )
    }
    // 函数计算结构的数据类型 输出数据的数据类型
    override def dataType: DataType = LongType
    //是否开启函数的稳定性 true
    override def deterministic: Boolean = true
    //缓冲区的初始化
    override def initialize(buffer: MutableAggregationBuffer): Unit = {
      //对 total count 进行赋值为0
      buffer.update(0,0L)//0 total
      buffer.update(1,0L)//1 count
    }
    //更新缓冲区
    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      buffer.update(0,buffer.getLong(0) + input.getLong(0) ) // total =  0 + 30
      buffer.update(1,buffer.getLong(1) + 1) // count = 0 + 1
    }
    //合并缓冲区
    override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit ={
      // 将 两个缓冲区的数据 进行相加 放到一个缓冲区当中
      buffer1.update( 0,buffer1.getLong(0) + buffer2.getLong(0)  )
      buffer1.update(1,buffer1.getLong(1)  +  buffer2.getLong(1) )
    }
    //将缓冲区里面的数据进行计算
    override def evaluate(buffer: Row): Any = {
      buffer.getLong(0)  / buffer.getLong(1)
    }
  }*/
}
